import com.github.brainlag.nsq.NSQConsumer;
import com.github.brainlag.nsq.NSQProducer;
import com.github.brainlag.nsq.lookup.LocalLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by root on 16-9-18.
 */
public class Main {
    private static Logger Log = LogManager.getLogger(Main.class);

    private static String randomString() {
        return "Message" + new Date().getTime();
    }

    public static void main(String[] args) throws Exception {
        int total = 10000 * 1000;

        AtomicInteger counter = new AtomicInteger(0);
        NSQLookup lookup = new LocalLookup(4150);

        NSQConsumer consumer = new NSQConsumer(lookup, "demo", "1", (message) -> {
            //Log.info("Processing message: " + new String(message.getMessage()));
            counter.incrementAndGet();
            message.finished();
        });
        consumer.start();

        NSQProducer producer = new NSQProducer();
        producer.addAddress("localhost", 4150);
        producer.start();
        String msg = randomString();
        for (int i = 0; i < total; i++) {
            producer.produce("demo", msg.getBytes());

            if( i % 10000 ==0){
                Log.info("send");
            }
        }
        Log.info("send done");
        producer.shutdown();

        while (counter.get() < total) {
            Thread.sleep(500);
        }

        Log.info("Sub Finished!");
        consumer.shutdown();
        Log.info("exit");
    }
}
